package skewreduce.framework;

import java.util.Collection;
import java.util.Comparator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import skewreduce.framework.physical.PhysicalOp;


public class PhysicalOpExecutor extends ThreadPoolExecutor {
	private final OperatorSchedule opSchedule;
	private final BlockingQueue<Future<SchedulerEvent>> completionQueue;
	private final CompletionService completionSvc;
	
	public PhysicalOpExecutor(EaggPlan plan,int corePoolSize,BlockingQueue<Future<SchedulerEvent>> q) {
		super(corePoolSize, corePoolSize, 0L, TimeUnit.MILLISECONDS, new PriorityBlockingQueue<Runnable>(corePoolSize,new TaskComparator()));
		opSchedule = plan.getSchedule();
		completionQueue = q;
		completionSvc = new CompletionService();
	}

	@Override
	protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
	    if (callable instanceof PhysicalOp)
            return new Task<T>(callable);
        else
        	throw new IllegalArgumentException();
	}

	@Override
	protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
		throw new IllegalArgumentException();
	}

	public CompletionService getCompletionService() {
		return completionSvc;
	}
	
    private final class Task<T> extends FutureTask<T> implements Comparable<Task<T>> {
    	private final PhysicalOp op;

		public Task(final Callable<T> tCallable) {
		    super(tCallable);
		    op = (PhysicalOp)tCallable;
		}
		
		@Override
		public int compareTo(final Task<T> o) {
			return opSchedule.compare(op,o.op);
		}
		
		@SuppressWarnings("unchecked")
		@Override
		protected void done() {
			completionQueue.add( (Future<SchedulerEvent>)(Task.this));
		}
    }

	private static class TaskComparator implements Comparator<Runnable> {
		@SuppressWarnings("unchecked")
		@Override
		public int compare(final Runnable left, final Runnable right) {
		    return ((Task) left).compareTo((Task) right);
		}
	}
	
	public final class CompletionService implements java.util.concurrent.CompletionService<SchedulerEvent> {
		@Override
		public Future<SchedulerEvent> poll() {
			return completionQueue.poll();
		}

		@Override
		public Future<SchedulerEvent> poll(long timeout, TimeUnit unit)
				throws InterruptedException {
			return completionQueue.poll(timeout, unit);
		}

		@Override
		public Future<SchedulerEvent> submit(Callable<SchedulerEvent> task) {
			if (task == null) throw new NullPointerException();
			return PhysicalOpExecutor.this.submit(task);
		}

		@Override
		public Future<SchedulerEvent> submit(Runnable task, SchedulerEvent result) {
			throw new UnsupportedOperationException();
		}

		@Override
		public Future<SchedulerEvent> take() throws InterruptedException {
			return completionQueue.take();
		}
		
        public int drainTo(Collection<? super Future<SchedulerEvent>> c) {
            return completionQueue.drainTo(c);
        }

        public int takeAndDrainTo(Collection<? super Future<SchedulerEvent>> c)
        throws InterruptedException {
            int n = completionQueue.drainTo(c);
            if ( n == 0 ) {
                Future<SchedulerEvent> e = completionQueue.take();
                c.add(e);
                n = completionQueue.drainTo(c) + 1;
            }
            return n;
        }

        public void shutdown(boolean force) {
            if ( force ) {
                PhysicalOpExecutor.this.shutdownNow();
            } else {
                PhysicalOpExecutor.this.shutdown();
            }
        }
	}
}
